package com.smile.sink;

import com.smile.beans.SensorReading;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.hadoop.shaded.org.apache.commons.httpclient.HttpHost;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;

//public class EsSink {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();;
//        DataStreamSource<String> dataStream = env.readTextFile("");
//        // es 的 httpHosts 配置
//        ArrayList<HttpHost> httpHosts = new ArrayList<>();
//        httpHosts.add(new HttpHost("localhost", 9200));
//        dataStream.addSink( new ElasticsearchSink.Builder<SensorReading>(httpHosts, new MyEsSinkFunction()).build());
//    }
//}
//public static class MyEsSinkFunction implements
//        ElasticsearchSinkFunction<SensorReading> {
//    @Override
//    public void process(SensorReading element, RuntimeContext ctx, RequestIndexer
//            indexer) {
//        HashMap<String, String> dataSource = new HashMap<>();
//        dataSource.put("id", element.getId());
//        dataSource.put("ts", element.getTimestamp().toString());
//        dataSource.put("temp", element.getTemperature().toString());
//        IndexRequest indexRequest = Requests.indexRequest()
//                .index("sensor")
//                .type("readingData")
//                .source(dataSource);
//        indexer.add(indexRequest);
//    }
//}

